package org.example.statistics
import java.text.SimpleDateFormat
import java.util.Date
import org.apache.spark.SparkConf
import org.apache.spark.sql.{DataFrame, SparkSession}

case class Rating(userId: Int, productId: Int, score: Double, timestamp: Int)

case class MongoConfig(uri: String, db: String)


object StatisticsRecommender {

  val MONGODB_RATING_COLLECTION = "Rating"
  //定义mongodb中统计后的存储表
  val RATE_MORE_PRODUCTS = "RateMoreProducts"
  val RATE_RECENTLY_PRODUCTS = "RateRecentlyProducts"
  val AVERAGE_PRODUCTS = "AverageProducts"

  def main(args: Array[String]): Unit = {

    val config = Map(
      "spark.cores" -> "local[*]",
      "mongo.uri" -> "mongodb://root:123456@192.168.175.10:27017/recommender?authSource=admin",
      "mongo.db" -> "recommender"
    )

    //创建spark config
    val sparkConf = new SparkConf().setMaster(config("spark.cores")).setAppName("StatisticsRecommender")
    //创建spark session
    val spark = SparkSession.builder().config(sparkConf).getOrCreate()

    import spark.implicits._
    implicit val mongoConfig = MongoConfig(config("mongo.uri"), config("mongo.db"))

    //加载数据
    val ratingDF = spark.read
      .option("uri", mongoConfig.uri)
      .option("collection", MONGODB_RATING_COLLECTION)
      .format("com.mongodb.spark.sql")
      .load()
      .as[Rating]
      .toDF()

    //创建rating临时表
    ratingDF.createOrReplaceTempView("ratings")

    //1.历史热门数据，按评分个数
    val rateMoreProductsDF = spark.sql("select productId, count(productId) as count from ratings group by productId order by count desc")
    storeDFInMongoDB(rateMoreProductsDF, RATE_MORE_PRODUCTS)


    //2.近期热门商品，把时间戳转换为yyyyMM格式进行评分个数统计
    //最终得到productId, count ,yearmonth
    //创建一个日期格式化工具
    val simpleDataFormat: SimpleDateFormat = new SimpleDateFormat("yyyyMM")
    //注册UDF，将timestamp转化为年月格式,将秒数转化为毫秒数再转化为年月格式
    spark.udf.register("changeDate", (x:Int) => simpleDataFormat.format(new Date(x * 1000L)).toInt)
    // 将原来的Rating数据集中的时间转换成年月的格式
    val ratingOfYearMonth = spark.sql("select productId, score, changeDate(timestamp) as yearmonth from ratings")
    // 将新的数据集注册成为一张表
    ratingOfYearMonth.createOrReplaceTempView("ratingOfMonth")
    //按照yearmonth与productId进行聚合，再按照yearmonth与count进行降序排序
    val rateMoreRecentlyProducts = spark.sql("select productId, count(productId) as count ,yearmonth from ratingOfMonth group by yearmonth,productId order by yearmonth desc, count desc")
    //把df保存到mongodb
    storeDFInMongoDB(rateMoreRecentlyProducts,RATE_RECENTLY_PRODUCTS)

    //3.优质商品统计，商品的平均评分
    val averageProducts = spark.sql("select productId, avg(score) as avg from ratings group by productId order by avg desc")
    storeDFInMongoDB( averageProducts, AVERAGE_PRODUCTS)

    spark.stop()
  }

  def storeDFInMongoDB(df: DataFrame, collectionName: String)(implicit mongoConfig: MongoConfig): Unit ={
    df.write
      .option("uri", mongoConfig.uri)
      .option("collection", collectionName)
      .mode("overwrite")
      .format("com.mongodb.spark.sql")
      .save()


  }
}